package com.gitee.xhs.joining.intervalJoin;

import com.gitee.xhs.OperatorSourceFactory;
import com.gitee.xhs.customFunction.OperatorKeyBy;
import com.gitee.xhs.customWaterMark.CustomWaterStrategy;
import com.gitee.xhs.dao.IntervalJoinDao;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author BigHead
 * @date 2022/7/21 9:40
 * @description IntervalJoinLogicMain
 * 这里的展示逻辑，mainStream每一条数据都会通过intervalJoin产生三条数据
 */
public class IntervalJoinMain {

    public SingleOutputStreamOperator<IntervalJoinDao> intervalJoinLogic(StreamExecutionEnvironment streamExecutionEnvironment, DataStream<Transaction> mainStream) {

        // IntervalJoin source
        KeyedStream<Transaction, Long> transactionKeyedStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTwentyOneSecondSource()
                )
                .map(
                        new OperatorIntervalMap()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                );

        return mainStream
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                )
                .intervalJoin(
                        transactionKeyedStream
                )
                .between(Time.milliseconds(-501), Time.milliseconds(501))
// 增加如下两个参数，则每一条mainStream只会匹配到一条数据
//                .upperBoundExclusive()
//                .lowerBoundExclusive()
                .process(
                        new OperatorIntervalJoin()
                );


    }

}
